/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月23日 下午3:33:22
*/
package cn.com.guage.flink.window;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @version:1.0.1
 * @Description: （对类进行功能描述）
 * @author: yangdechao
 * @date: datedate 2021年11月23日 下午3:33:22
 */
public class TumblingProcessingTimeWindowsTest2 {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		// 如果不指定时间的话，默认是 ProcessingTime，但是如果指定为事件事件的话，需要事件中带有时间或者添加时间水印
//	    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		/**
		 * 输入数据格式如: my name is yangdechao
		 * 
		 */
		DataStreamSource<String> data = env.socketTextStream("192.168.0.128", 9528);
		data.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public void flatMap(String in, Collector<Tuple2<String, Integer>> out) throws Exception {
				String[] splits = in.split(" ");
				for (String word : splits) {
					out.collect(new Tuple2<String, Integer>(word, 1));
				}
			}
		}).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = -5236084965539684057L;

			@Override
			public String getKey(Tuple2<String, Integer> value) throws Exception {
				return value.f0;
			}

		}).window(TumblingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(3))).sum(1).print();//滚动时间窗口大小为5秒,但是
		env.execute("SessionWindowsTest");
	}
}
